package com.viewscenes.netsupervisor.netty.server;

import com.viewscenes.netsupervisor.annotation.RpcService;
import com.viewscenes.netsupervisor.netty.codec.json.JSONDecoder;
import com.viewscenes.netsupervisor.netty.codec.json.JSONEncoder;
import com.viewscenes.netsupervisor.registry.ServiceRegistry;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.Map;

/**
 * @program: rpc-provider
 * @description: ${description}
 * @author: yyp
 * @create: 2018-11-30 17:10
 **/
@Component
public class NettyServer implements ApplicationContextAware,InitializingBean{

    private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);
    private static final EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    private static final EventLoopGroup workerGroup = new NioEventLoopGroup(4);

    private Map<String, Object> serviceMap = new HashMap<>();

    @Value("${rpc.server.address}")
    private String serverAddress;

    @Autowired
    ServiceRegistry registry;

    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {

        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(RpcService.class);
        for(Object serviceBean:beans.values()){

            Class<?> clazz = serviceBean.getClass();

            Class<?>[] interfaces = clazz.getInterfaces();

            for (Class<?> inter : interfaces){
                String interfaceName = inter.getName();
                logger.info("加载服务类: {}", interfaceName);
                serviceMap.put(interfaceName, serviceBean);
            }
        }
        logger.info("已加载全部服务接口:{}", serviceMap);
    }

    public void afterPropertiesSet() throws Exception {
        start();
    }

    /**
     * EventLoop里面关键属性有两个，多路复用器Selector和任务队列。可以把通道（Channel）注册在多路复用器上面，可以不断轮询其中的事件然后执行。任务队列存储提交的task。
     * EventLoop处理的事件（叫任务也行，事件更加贴切吧）整体上有两种：
     * IO事件。当一个EventLoop所关联的多路复用器上面注册的通道发生“连接、接收（Acceptor）、读、写”事件的时候，就相当于触发了IO事件。一般也就两种场景：作为server端的时候，监听一个端口，别人来访问你的端口，就会先触发接收事件，然后读取，写入事件。作为client端的时候，要和目标连接，连接成功以后就会触发连接事件，然后写入，读取事件。（场景简化了一下）
         * 非IO事件。这边又分为两种：
         * 普通任务。使用execute提交的任务，直接执行的。
         * 调度任务。使用schedule提交的任务，一般需要延迟或者周期性执行的。
     *
     * EventLoop在执行的时候，也是无线循环，循环体内主要有3件事：阻塞轮询、执行IO事件和执行非IO事件。
        *     若当前没有任务非IO事件（普通任务）需要执行，且在0.5s内没有需要执行的调度任务的时候，先会进入一个无限循环，里面会调用多路复用器的select(long)方法进行阻塞超时轮询，阻塞超时默认是1s或者有定时任务的话，就取定时任务应该执行的时间与当前时间的间隔为超时时间（意思就是，我超时结束的时候，最早的定时任务刚好可以执行了）。
        *      多路复用器的阻塞超时轮询，并不会一直等到超时，有多种方式可以唤醒它：
                 * 多路复用器已经准备好了至少一个事件；基本上就是有IO事件的话，就直接返回了，不会阻塞。
                 * 使用wakeup方法。当其它线程调用的时候，会立刻唤醒正在阻塞轮询多路复用器的线程。而EventLoop也是利用了这一点，当有新的任务提交进来，并且当前情况满足4个条件的话，就会执行wakeUp。条件很好满足。而且其中某些条件就是在判断是不是在做阻塞轮询，如果是的话，才会去唤醒。
                 * 当正在阻塞轮询的时候，有新的非IO任务进来的话，就会立刻唤醒。和上一点是一回事，换了一种说法。
                 * 这边也有一个骚操作，它在执行一些中断操作的时候，会提交一个空任务来唤醒。
                 * 超时时间到。
                 * 当前线程被中断。后面这两种没有什么可说的。
     * 它的这种唤醒机制，保证了不会影响到任何事件。但是仔细想想，这也是应该的，毕竟是它实在没有事情做的时候，才回去阻塞轮询，因为对于NIO而已，根本不需要进行阻塞，你去忙你的，忙完了回来叫我，我都给你准备好了，你忙你的，我做我的，相互不影响（你=eventLoop,我=多路复用器）。正因为如此，它的代码实现上面，对于跳出无限阻塞轮询（阻塞轮询外层有个无限循环）的条件也是非常开放（不知道怎么描述了），很容易就跳出了，可以看看代码。
     * 阻塞轮询完了或者根本不需要阻塞轮询的（有非IO事件），就要处理事件了。它这边有个IO比例，默认是50，就是IO:非IO=50:50，比如处理IO的时间是100ms，那么处理非IO的时间最大也得是100ms,但是它并没有强行去限制，也确实不好做。它仅仅只是在每执行64个非IO事件以后去判断一下这个时间，超了的话，就停下来。64，也不知道是怎么定义的，说实话我觉得挺多的，太小的话，是不是就会影响到非IO任务的执行了呢？还有这个IO比例，当=100的时候，就完全忽略了时间比，每轮询一次，就会把剩余的所有非IO全部执行完。既然都是IO比例了，这种情况就不应该是只执行IO吗？只执行IO肯定不对，但是这个实现和对应的情况实在是不搭呀，理解不了。或许是因为有些事情我还没有理解透彻。
     * 执行IO的时候，就是把所有轮询到的事件，挨个去执行。这块就是我开篇提到的第四个核心，不过我还没有细看（主要是挺复杂的，不花点事情是搞不明白的），就不说了。反正是一个一个执行IO事件，而且肯定是用当前线程去执行，但是肯定不会花太多时间去处理完的，到最后一定会交给另外一个EventLoopGroup，这也是标准的Reactor模型。netty服务端启动的时候，需要提供两个EventLoopGroup，也是这个作用吧，我猜的。
     * 执行非IO的时候，先把调度队列中所有到期的取出来放进任务队列中，然后挨个去执行。一个是全部执行完，一个有时间限制。执行完了以后，会执行tailTasks队列里面的任务，这个设计不知道用来干嘛的，意思就是每一次轮询结束，就去执行一下。感觉没有什么用呀。
     * 结束以后，下一波轮询又开始了
     */
    public void start(){

        final NettyServerHandler handler = new NettyServerHandler(serviceMap);

        new Thread(() -> {
            try {
                ServerBootstrap bootstrap = new ServerBootstrap();
                //  bootstrap.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class) 相当于 nio的 channel绑定到selector上
                // 0.0 是初始化两个线程，一个线程负责接受新的连接，一个负责处理读写；是开启了；
                // 1.0   bossGroup线程的机制是多路复用，虽然是一个线程但是可以监听多个新连接；
                    //举个例子：
                    // 假设有A,B,C三个连接，先连接A，如果A还没连接上，立刻返回，不会一直等待，就不会阻塞其他客户端连接服务器；
                    //  然后去连接B，B连接上了；
                    //  再连接C，C连接还没到的话，立刻返回，不去等待，避免其他客户端连接不上阻塞；
                    // 现在再返回连接A，A连接上了，就ok了；
                    //核心思想：轮询，没有连接立刻返回不要耽误其他的客户端连接，不会阻塞；一个线程不会浪费资源；
                //2.0 这个代码怎么与多路复用selector怎么挂钩？
                    // 解惑：两个线程，一个是bossGroup，一个是workerGroup，他们的机制都是多路复用selector，一个线程监听多个客户端，
                    // 看上面问题的例子，就知道什么是多路复用selector
                //3、bossGroup负责与客户端建立连接的，既然有客户端等待连接，就不存在连不上，怎么会需要多路复用轮询呢？
                    //解惑： 需要用多路复用，服务器与客户端一对多，轮询的核心代码是for循环；既然存在客户端连接，不存在连接不上，轮询次数比较少（除非客户端有问题）
                bootstrap.group(bossGroup,workerGroup).
                        channel(NioServerSocketChannel.class).
                        option(ChannelOption.SO_BACKLOG,1024).
                        childOption(ChannelOption.SO_KEEPALIVE,true).
                        childOption(ChannelOption.TCP_NODELAY,true).
                        childHandler(new ChannelInitializer<SocketChannel>() {
                            //创建NIOSocketChannel成功后，在进行初始化时，将它的ChannelHandler设置到ChannelPipeline中，用于处理网络IO事件
                            protected void initChannel(SocketChannel channel) throws Exception {
                                ChannelPipeline pipeline = channel.pipeline();
                                pipeline.addLast(new IdleStateHandler(0, 0, 60));
                                pipeline.addLast(new JSONEncoder());
                                pipeline.addLast(new JSONDecoder());
                                pipeline.addLast(handler);
                            }
                        });

                String[] array = serverAddress.split(":");
                String host = array[0];
                int port = Integer.parseInt(array[1]);
                ChannelFuture cf = bootstrap.bind(host,port).sync();
                logger.info("RPC 服务器启动.监听端口:"+port);
                registry.register(serverAddress);
                //等待服务端监听端口关闭
                cf.channel().closeFuture().sync();
            } catch (Exception e) {
                e.printStackTrace();
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }).start();
    }
}
